#include "TcpSocketServer.h"
#include <assert.h>

ConnNode::ConnNode(int fd) : m_fd(fd)
{
	m_WriteBuf = NULL;
	m_ReadBuf = NULL;
	m_Prev = NULL;
	m_Next = NULL;
}

ConnNode::~ConnNode()
{

}

ConnQueue::ConnQueue()
{
	//建立头尾结点，并调整其指针
	m_head = new ConnNode(0);
	m_tail = new ConnNode(0);
	m_head->m_Prev = m_tail->m_Next = NULL;
	m_head->m_Next = m_tail;
	m_tail->m_Prev = m_head;
	//m_NodeCount = 0;
}

ConnQueue::~ConnQueue()
{
	ConnNode *tcur, *tnext;
	tcur = m_head;
	//循环删除链表中的各个结点
	while (tcur != NULL)
	{
		tnext = tcur->m_Next;
		delete tcur;
		tcur = tnext;
	}
}

ConnNode* ConnQueue::InsertConnNode(evutil_socket_t fd, LibeventThread *t)
{
	if (m_NodeCount+1 > MAX_ACCEPT_NUM)
		return NULL;
	m_NodeCount++;
	ConnNode *c = new ConnNode(fd);
	c->m_Thread = t;
	ConnNode *next = m_head->m_Next;

	c->m_Prev = m_head;
	c->m_Next = m_head->m_Next;
	m_head->m_Next = c;
	next->m_Prev = c;
	return c;
}

void ConnQueue::DeleteConnNode(ConnNode *c)
{
	if (NULL != c)
	{
		c->m_Prev->m_Next = c->m_Next;
		c->m_Next->m_Prev = c->m_Prev;
		delete c;
		m_NodeCount--;
	}
}

void ConnQueue::PrintQueue()
{
	ConnNode *cur = m_head->m_Next;
	while( cur->m_Next != NULL )
	{
		printf("%d ", cur->m_fd);
		cur = cur->m_Next;
	}
	printf("\n");
}


TcpSocketServer::TcpSocketServer(int threadCount)
{
	//初始化各项数据
	m_ThreadCount = threadCount;
	m_Port = EXIT_CODE;
	m_MainBase = new LibeventThread;
	if(threadCount > 0)
	{
		m_Threads = new LibeventThread[m_ThreadCount];
	}
	//多线程支持
#ifdef WIN32
		evthread_use_windows_threads();
#else
        evthread_use_pthreads();
#endif
#ifdef WIN32
        m_MainBase->tid = GetCurrentThreadId();
#else
        m_MainBase->tid = pthread_self();
#endif
	m_MainBase->tcpConnect = this;
	m_MainBase->base = event_base_new();
	//printf("MainBase=%x",m_MainBase->base);

	enum event_method_feature f;
	f = (enum event_method_feature)event_base_get_features(m_MainBase->base);
	if((f & EV_FEATURE_ET))
        printf(" Edge-triggered events are supported.\n");
    if((f & EV_FEATURE_O1))
        printf(" O(1) events notification are supported.\n");
    if((f & EV_FEATURE_FDS))
        printf(" All FD types are supported.\n");

	//初始化各个子线程的结构体
	for (int i = 0; i < m_ThreadCount; i++)
	{
		SetupThread(&m_Threads[i]);
	}
}

TcpSocketServer::~TcpSocketServer()
{
	//停止事件循环（如果事件循环没开始，则没效果）
	StopRun(NULL);

	//释放内存
	event_base_free(m_MainBase->base);
	for (int i = 0; i<m_ThreadCount; i++)
		event_base_free(m_Threads[i].base);

	delete m_MainBase;
	delete[] m_Threads;
}

void TcpSocketServer::ErrorQuit(const char *str)
{
	//输出错误信息，退出程序
	fprintf(stderr, "%s", str);
	if (errno != 0)
		fprintf(stderr, " : %s", strerror(errno));
	fprintf(stderr, "\n");
	exit(1);
}
void TcpSocketServer::SetupThread(LibeventThread *me)
{
	int ret = 0;
	//建立libevent事件处理机制
	me->tcpConnect = this;
	me->base = event_base_new();
	if (NULL == me->base)
		ErrorQuit("event base new error");

	//在主线程和子线程之间建立管道
	evutil_socket_t fd[2];
#ifdef WIN32
	//ret = evutil_socketpair(AF_INET, SOCK_STREAM, 0, fd);
	ret = evutil_socketpair(LOCAL_SOCKETPAIR_AF, SOCK_STREAM, 0, fd);
#else
	ret = evutil_socketpair(AF_UNIX, SOCK_STREAM, 0, fd);
#endif
	if(ret != 0)
		ErrorQuit("evutil_socketpair");
	me->notifyReceiveFd = fd[0];
	me->notifySendFd = fd[1];
	//让子线程的状态机监听管道
	event_set(&me->notifyEvent, me->notifyReceiveFd,
		EV_READ | EV_PERSIST, ThreadProcess, me);
	event_base_set(me->base, &me->notifyEvent);
	if (event_add(&me->notifyEvent, 0) == -1)
		ErrorQuit("Can't monitor libevent notify pipe\n");
}

#ifdef WIN32
unsigned int WINAPI TcpSocketServer::WorkerLibevent(LPVOID inThread)
#else
void* TcpSocketServer::WorkerLibevent(void *inThread)  //static
#endif
{
	//开启libevent的事件循环，准备处理业务
	LibeventThread *me = (LibeventThread*)inThread;
	//printf("thread %u started\n", (unsigned int)me->tid);
	event_base_dispatch(me->base);
	printf("TcpSocketServer's child thread %llu is over\n",me->tid);
	return NULL;
}

bool TcpSocketServer::StartRun()
{
	evconnlistener *listener;

	//如果端口号不是EXIT_CODE，就监听该端口号
	if (m_Port != EXIT_CODE)
	{
		sockaddr_in sin;
		memset(&sin, 0, sizeof(sin));
		sin.sin_family = AF_INET;
		sin.sin_port = htons(m_Port);
		listener = evconnlistener_new_bind(m_MainBase->base,
			ListenerEventCb, (void*)this,
			LEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE | LEV_OPT_THREADSAFE, -1,
			(sockaddr*)&sin, sizeof(sockaddr_in));
		if (NULL == listener)
			ErrorQuit("TCP listen error");
	}
	else
	{
        fprintf(stderr,"TCPServer's Port is not be assigned.\n");
		return false;
	}

	//开启各个子线程
	for (int i = 0; i < m_ThreadCount; i++)
	{
#ifdef WIN32
		unsigned int theId = 0; // We don't care about the identifier
		m_Threads[i].tid = (HANDLE)_beginthreadex(NULL,   // Inherit security
			0,      // Inherit stack size
			WorkerLibevent,			 // Entry function
			(void*)&m_Threads[i],    // Entry arg
			0,						 // Begin executing immediately
			&theId);
		assert(m_Threads[i].tid != NULL);
#else
		int err = pthread_create(&m_Threads[i].tid, NULL,
			WorkerLibevent, (void*)&m_Threads[i]);
		assert(err == 0);
#endif
	}

	//开启主线程的事件循环
	event_base_dispatch(m_MainBase->base);

	//事件循环结果，释放监听者的内存
	if (m_Port != EXIT_CODE)
	{
		//printf("free listen\n");
		evconnlistener_free(listener);
	}
	return true;
}
void TcpSocketServer::StopRun(timeval *tv)
{
	int contant = EXIT_CODE;
	//向各个子线程的管理中写入EXIT_CODE，通知它们退出
	for (int i = 0; i<m_ThreadCount; i++)
	{
		send(m_Threads[i].notifySendFd, (char*)&contant, sizeof(int),0);
	}
	//结束主线程的事件循环
	event_base_loopexit(m_MainBase->base, tv);
    printf("TcpServer has been stoped!");
}

void TcpSocketServer::ListenerEventCb(struct evconnlistener *listener,
	evutil_socket_t fd,
	struct sockaddr *sa,
	int socklen,
	void *user_data)
{
	TcpSocketServer *server = (TcpSocketServer*)user_data;

	if (server->m_ThreadCount > 0)
	{
		//随机选择一个子线程，通过管道向其传递socket描述符
		int num = rand() % server->m_ThreadCount;
		evutil_socket_t sendfd = server->m_Threads[num].notifySendFd;
		send(sendfd, (char*)&fd, sizeof(evutil_socket_t),0);
	}
	else
	{
		LibeventThread *me = server->m_MainBase;
		//新建连接
		struct bufferevent *bev;
		// 设置非阻塞
        evutil_make_socket_nonblocking(fd);
        // 如果调用了exec,应该关闭指定的套接字
        evutil_make_socket_closeonexec(fd);
		//printf("MainBase=%x",me->base);

		bev = bufferevent_socket_new(me->base, fd, BEV_OPT_CLOSE_ON_FREE | BEV_OPT_THREADSAFE);
		if (!bev)
		{
			fprintf(stderr, "bufferevent_socket_new error:Error constructing bufferevent!");
			evutil_closesocket(fd);
            //主线程监听客户端时，没有必要退出循环，继续等待下一个客户端
		//	event_base_loopbreak(me->base);
			return;
		}

		//将该链接放入队列
		ConnNode *conn = me->connectQueue.InsertConnNode(fd, me);
		conn->m_ReadBuf = bufferevent_get_input(bev);
		conn->m_WriteBuf = bufferevent_get_output(bev);
		if (conn == NULL)
		{
			fprintf(stderr, "Error client is full!");
			struct evbuffer* output;
			output = bufferevent_get_output(bev);
			evbuffer_add(output, "Server is full!", sizeof("Server is full!"));
			bufferevent_free(bev);
			evutil_closesocket(fd);
			return;
		}
		//准备从socket中读写数据
		bufferevent_setcb(bev, ReadEventCb, WriteEventCb, CloseEventCb, conn);
		bufferevent_enable(bev, EV_READ|EV_WRITE);

		//调用用户自定义的连接事件处理函数
		me->tcpConnect->ConnectionEvent(conn);

	}
}
void TcpSocketServer::ThreadProcess(int fd, short which, void *arg)
{
	LibeventThread *me = (LibeventThread*)arg;

	//从管道中读取数据（socket的描述符或操作码）
	evutil_socket_t pipefd = me->notifyReceiveFd;
	evutil_socket_t connfd;
	recv(pipefd, (char*)&connfd, sizeof(evutil_socket_t),0);

	//如果操作码是EXIT_CODE，则终于事件循环
	if (EXIT_CODE == connfd)
	{
		event_base_loopbreak(me->base);
        // 关闭客户端的文件描述符
		evutil_closesocket(connfd);
		return;
	}
	printf("New ThreadId=%p\n", &me->tid);
	//新建连接
	struct bufferevent *bev;
	// 设置非阻塞
    evutil_make_socket_nonblocking(connfd);
    // 如果调用了exec,应该关闭指定的套接字
    evutil_make_socket_closeonexec(connfd);
    int bufSize = 1024 * 200;
    ::setsockopt(connfd, SOL_SOCKET, SO_SNDBUF, (char*)&bufSize, sizeof(int));
    int one = 1;
    setsockopt(connfd,IPPROTO_TCP,TCP_NODELAY,&one,sizeof one);
	bev = bufferevent_socket_new(me->base, connfd, BEV_OPT_CLOSE_ON_FREE | BEV_OPT_THREADSAFE);
	if (!bev)
	{
		fprintf(stderr, "bufferevent_socket_new:Error constructing bufferevent!");
		evutil_closesocket(connfd);
		event_base_loopbreak(me->base);
		return;
	}
	//bufferevent_setwatermark(bev,EV_WRITE,0,10000);

	//将该链接放入队列
	ConnNode *conn = me->connectQueue.InsertConnNode(connfd, me);
	if (conn == NULL)
	{
		fprintf(stderr, "Error client is full!");
		struct evbuffer* output;
		output = bufferevent_get_output(bev);
		evbuffer_add(output, "Server is full!", sizeof("Server"));
		bufferevent_free(bev);
		evutil_closesocket(connfd);
		event_base_loopbreak(me->base);
		return;
	}
	conn->m_ReadBuf = bufferevent_get_input(bev);
	conn->m_WriteBuf = bufferevent_get_output(bev);
	evbuffer_enable_locking(conn->m_ReadBuf,NULL);
	evbuffer_enable_locking(conn->m_WriteBuf,NULL);
	//bufferevent_set_max_single_write(bev,16384);
	
	//bufferevent_set_max_single_write(bev,66535);
	//ev_ssize_t maxwrite = bufferevent_get_max_single_write(bev);
	//printf("maxwrite=%d",maxwrite);
	//准备从socket中读写数据
	bufferevent_setcb(bev, ReadEventCb, WriteEventCb, CloseEventCb, conn);
	bufferevent_enable(bev, EV_READ|EV_WRITE);

	//调用用户自定义的连接事件处理函数
	me->tcpConnect->ConnectionEvent(conn);

}

void TcpSocketServer::ReadEventCb(struct bufferevent *bev, void *data)
{
	//bufferevent_lock(bev);
	ConnNode *conn = (ConnNode*)data;
	//conn->m_ReadBuf = bufferevent_get_input(bev);
	//conn->m_WriteBuf = bufferevent_get_output(bev);


	//调用用户自定义的读取事件处理函数
	conn->m_Thread->tcpConnect->ReadEvent(conn);
	//bufferevent_unlock(bev);
}

void TcpSocketServer::WriteEventCb(struct bufferevent *bev, void *data)
{
	//bufferevent_lock(bev);
	ConnNode *conn = (ConnNode*)data;
	//conn->m_ReadBuf = bufferevent_get_input(bev);
	//conn->m_WriteBuf = bufferevent_get_output(bev);

	//调用用户自定义的写入事件处理函数
	conn->m_Thread->tcpConnect->WriteEvent(conn);
	//bufferevent_unlock(bev);
}

void TcpSocketServer::CloseEventCb(struct bufferevent *bev, short events, void *data)
{
	//bufferevent_lock(bev);
	ConnNode *conn = (ConnNode*)data;
	//调用用户自定义的断开事件处理函数
	conn->m_Thread->tcpConnect->CloseEvent(conn, events);
	conn->GetThread()->connectQueue.DeleteConnNode(conn);
	//bufferevent_unlock(bev);
	bufferevent_free(bev);
}

bool TcpSocketServer::AddSignalEvent(int sig, void(*ptr)(int, short, void*))
{
	//新建一个信号事件
	/*
	event *ev = evsignal_new(m_MainBase->base, sig, ptr, (void*)this);
	if (!ev ||
		event_add(ev, NULL) < 0)
	{
		event_del(ev);
		return false;
	}

	//删除旧的信号事件（同一个信号只能有一个信号事件）
	DeleteSignalEvent(sig);
	m_SignalEvents[sig] = *ev;
	*/
	return true;
}

bool TcpSocketServer::DeleteSignalEvent(int sig)
{
	//event* ev = &(m_SignalEvents.get(sig));
	//if(ev == NULL)
	//	return false;

	//event_del(ev);
	//m_SignalEvents.remove(sig);
	return true;
}

event *TcpSocketServer::AddTimerEvent(void(*ptr)(int, short, void *),
	timeval tv, bool once)
{
	int flag = 0;
	if (!once)
		flag = EV_PERSIST;

	//新建定时器信号事件
	event *ev = new event;
	event_assign(ev, m_MainBase->base, -1, flag, ptr, (void*)this);
	if (event_add(ev, &tv) < 0)
	{
		event_del(ev);
		return NULL;
	}

	/*struct event* ev;
	ev = event_new(m_MainBase->base,-1,flag,ptr,(void*)this);
	if(event_add(ev,&tv)< 0)
	{
		event_del(ev);
		return NULL;
	}*/

	return ev;
}

bool TcpSocketServer::DeleteTimerEvent(event *ev)
{
	int res = event_del(ev);
	return (0 == res);
}
